//#pragma once

#include <iostream>
#include <vector>
#include <string>
#include <unordered_map>
#include <functional>
#include <memory>
#include <cstring>
#include <assert.h>
#include <algorithm>
#include <ctime>
#include <signal.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/epoll.h>
#include <sys/eventfd.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <fcntl.h>
#include <mutex>
#include <thread>
#include <cstdint>
#include <sys/timerfd.h>
#include <typeinfo>
#include <condition_variable>


#define INF 0
#define DBG 1
#define ERR 2
#define LOG_LEVEL ERR  //控制日志等级

#define LOG(level, format, ...) do{\
        if (level < LOG_LEVEL) break;\
        time_t t = time(NULL);\
        struct tm *ltm = localtime(&t);\
        char tmp[32] = {0};\
        strftime(tmp, 31, "%H:%M:%S", ltm);\
        fprintf(stdout, "[%p %s %s:%d] " format "\n", (void*)pthread_self(), tmp, __FILE__, __LINE__, ##__VA_ARGS__);\
    }while(0)

#define INF_LOG(format, ...) LOG(INF, format, ##__VA_ARGS__)
#define DBG_LOG(format, ...) LOG(DBG, format, ##__VA_ARGS__)
#define ERR_LOG(format, ...) LOG(ERR, format, ##__VA_ARGS__)

#define BUFFER_DEFAULT_SIZE 1024
class Buffer
{
    private:
        std::vector<char> _buffer; //使用vector进行空间管理
        uint64_t _reader_idx; //读偏移
        uint64_t _writer_idx; //写偏移
    public:
        Buffer():_reader_idx(0), _writer_idx(0), _buffer(BUFFER_DEFAULT_SIZE){}
         char* Begin() {return &*_buffer.begin(); }
        //获取当前写入起始位置:_buffer的空间起始地址+写偏移量
         char* WritePos(){ return Begin() + _writer_idx; }
        //获取当前读取起始地址
         char* ReadPos() {return Begin() + _reader_idx;}
        //获取缓冲区末尾空闲空间大小--写偏移之后的空闲空间:总空间-写偏移
        uint64_t TailIdleSize() {return _buffer.size() - _writer_idx; }
        //获取缓冲区起始空闲空间大小--读偏移之前的空闲空间: 已经读完的空间，就是空闲的空间
        uint64_t HeadIdleSize() { return _reader_idx; }
        //获取可读数据大小: 写偏移 - 读偏移
        uint64_t ReadAbleSize() { return _writer_idx - _reader_idx; }
        //将读偏移向后移动
        void MoveReadOffset(uint64_t len) { 
            if(len == 0) return;
            //向后移动的大小，必须小于可读数据的大小
            assert(len <= ReadAbleSize());
            _reader_idx += len;
        }
        //将写偏移向后移动
        void MoveWriteOffset(uint64_t len){
            //向后移动的大小，必须小于当前后面的空闲空间大小
            assert(len <= TailIdleSize());
            _writer_idx += len;
        }
        //确保写空间足够(整体空间够了就移动数据，否则就扩容)
        void EnsureWriteSpace(uint64_t len){
            //如果后面空闲空间大小足够，直接返回
            if(TailIdleSize() >= len){ return; }
            //后面空闲空间大小不够，则判断加上前面空闲空间大小是否足够，
            //若足够，则将数据往前挪，否则就扩容
            if(len <= TailIdleSize() + HeadIdleSize())
            {
                uint64_t rsz = ReadAbleSize(); //保存当前数据大小
                std::copy(ReadPos(), ReadPos() + rsz, Begin()); //把当前数据拷贝到起始位置
                _reader_idx = 0;
                _writer_idx = rsz;
            }else{
                //剩余的所有空间都不够，需要扩容，不移动数据，直接在写偏移之后扩容足够的空间
                _buffer.resize(_writer_idx + len);
            }
        }
        //写入数据
        void Write(const void* data, uint64_t len){
            //1.保证有足够空间 2.拷贝数据进去
            EnsureWriteSpace(len);
            const char* d = (const char*)data;
            std::copy(d, d + len, WritePos());
        }
        void WriteAndPush(void* data, uint64_t len)
        {
            Write(data, len);
            MoveWriteOffset(len);
        }
        //针对字符串的写入
        void WriteString(const std::string& data)
        {
            return Write((void*)data.c_str(), data.size());
        }
        void WriteStringAndPush(const std::string& data)
        {
            WriteString(data);
            MoveWriteOffset(data.size());
        }
        //将另一个缓冲区的数据写入到这个缓冲区
        void WriteBuffer(Buffer& data)
        {
            return Write((void*)data.ReadPos(), data.ReadAbleSize());
        }
        void WriteBufferAndPush(Buffer& data)
        {
            WriteBuffer(data);
            MoveWriteOffset(data.ReadAbleSize());
        }
        //读入数据
        void Read(void* buf, uint64_t len){
            //要求获取的数据大小必须小于可读的数据大小
            assert(len <= ReadAbleSize());
            std::copy(ReadPos(), ReadPos() + len, (char*)buf);
        }
        void ReadAndPop(void* buf, uint64_t len)
        {
            Read(buf, len);
            MoveReadOffset(len);
        }
        //把读取的数据当做字符串返回
        std::string ReadAsString(uint64_t len)
        {
            //要求获取的数据大小必须小于可读的数据大小
            assert(len <= ReadAbleSize());
            std::string str;
            str.resize(len);
            Read((void*)str.c_str(), len);
            return str;
        }
        std::string ReadAsStringAndPop(uint64_t len)
        {
            assert(len <= ReadAbleSize());
            std::string str = ReadAsString(len);
            MoveReadOffset(len);
            return str;
        }
        //找出回车换行\n
        char* FindCRLF(){
             void* res = memchr(ReadPos(), '\n', ReadAbleSize());
            return (char*)res;
        }
        //获取一行数据
        std::string GetLine(){
            char* pos = FindCRLF();
            if(pos == NULL){
                return "";
            }
            //+1是为了把换行字符也取出来
            return ReadAsString(pos - ReadPos() + 1);
        }
        std::string GetLineAndPop()
        {
            std::string str = GetLine();
            MoveReadOffset(str.size());
            return str;
        }
        //清空缓冲区
        void Clear(){
            //将偏移量都归0，下次从头开始覆盖写入即可
            _reader_idx = 0;
            _writer_idx = 0;
        }
};

#define MAX_LISTEN 1024
class Socket
{
    private:
        int _sockfd;
    public:
        Socket():_sockfd(-1){}
        Socket(int fd):_sockfd(fd){}
        ~Socket(){Close(); }
        int Fd() { return _sockfd; }
        //创建套接字
        bool Create(){
            //int socket(int domain, int type, int protocol)
            //            地址域类型   套接字类型     协议类型
            _sockfd = socket(AF_INET/*IPV4地址类型*/, SOCK_STREAM/*流式套接字*/, IPPROTO_TCP);
            if(_sockfd < 0){
                ERR_LOG("create socket fail!");
                return false;
            }
            return true;
        }
        //绑定地址信息
        bool Bind(const std::string& ip, uint16_t port){
            struct sockaddr_in addr;
            addr.sin_family = AF_INET;
            addr.sin_port = htons(port);
            addr.sin_addr.s_addr = inet_addr(ip.c_str());
            socklen_t len = sizeof(struct sockaddr_in);
            //int bind(int sockfd, struct sockaddr* addr, socklen_t len)
            int ret = bind(_sockfd, (struct sockaddr*)&addr, len);
            if(ret < 0)
            {
                ERR_LOG("bind server fail!");
                return false;
            }
            return true;
        }
        //开始监听
        bool Listen(int backlog = MAX_LISTEN){
            //int listen(int sockfd, int backlog)
            int ret = listen(_sockfd, backlog);
            if(ret < 0)
            {
                ERR_LOG("listen fail!");
                return false;
            }
            return true;
        }
        //向服务器发起连接
        bool Connect(const std::string& ip, uint16_t port){
            struct sockaddr_in addr;
            addr.sin_family = AF_INET;
            addr.sin_port = htons(port);
            addr.sin_addr.s_addr = inet_addr(ip.c_str());
            socklen_t len = sizeof(struct sockaddr_in);
            //int connect(int sockfd, struct sockaddr* addr, socklen_t len)
            int ret = connect(_sockfd, (struct sockaddr*)&addr, len);
            if(ret < 0)
            {
                ERR_LOG("connect server fail!");
                return false;
            }
            return true;
        }
        //获取新连接
        int Accept(){
            //int accept(int sockfd, struct sockaddr* addr, socklen_t len)
            int newfd = accept(_sockfd, NULL, NULL);
            if(newfd < 0){
                ERR_LOG("accept fail!");
                return -1;
            }
            return newfd;
        }
        //接收数据
        //ssize_t 表示有符号整形
        ssize_t Recv(void* buf, size_t len, int flag = 0){
            //ssize_t recv(int sockfd, void* buf, size_t len, int flag)
            ssize_t ret = recv(_sockfd, buf, len, flag);
            if(ret <= 0){ //等于0表示连接断开了
                //EAGAIN 表示当前socket的接收缓冲区中没有数据了，在非阻塞的情况下才会有这个错误
                //EINTR  表示当前socket的阻塞等待被信号打断了
                if(errno == EAGAIN || errno == EINTR){
                    return 0; //表示这次接收没有接收到数据
                }
                ERR_LOG("recv fail!");
                return -1;
            }
            return ret; //实际接收的数据长度
        }
        ssize_t NonBlockRecv(void* buf, size_t len){
            return Recv(buf, len, MSG_DONTWAIT); //MSG_DONTWAIT 表示当前接收为非阻塞(没有数据就报错返回，不会一直停在那里)
        }
        //发送数据
        ssize_t Send(const void* buf, size_t len, int flag = 0){
            //ssize_t send(int sockfd, void* buf, size_t len, int flag)
            int ret = send(_sockfd, buf, len, flag);
            if(ret < 0){
                if (errno == EAGAIN || errno == EINTR) {
                    return 0;
                }
                ERR_LOG("send fail!");
                return -1;
            }
            return ret; //实际发送的数据长度
        }
         ssize_t NonBlockSend(void* buf, size_t len){
            if(len == 0) return 0;
            return Send(buf, len, MSG_DONTWAIT); //MSG_DONTWAIT 表示当前发送为非阻塞(没有数据就报错返回，不会一直停在那里)
        }
        //关闭套接字
        void Close(){
            if(_sockfd != -1){
                close(_sockfd);
                _sockfd = -1;
            }
        }
        //创建一个服务器连接
        bool CreateServer(uint16_t port, const std::string& ip = "0.0.0.0", bool block_flag = false){
            //1.创建套接字 2.绑定套接字 3.监听套接字 4.设置为非阻塞 5.启动地址重用
            if(Create() == false) return false;
            ReuseAddress();
            if(block_flag) NonBlock();
            if(Bind(ip, port) == false) return false;
            if(Listen() == false) return false;
            return true;
        }
        //创建一个客户端连接
        bool CreateClient(uint16_t port, const std::string& ip){
            //1.创建套接字 2.直接连接服务器
            if(Create() == false) return false;
            if(Connect(ip, port) == false) return false;
            return true;
        }
        //设置套接字选项--开启地址端口重用
        //一个连接绑定了ip和端口之后，一旦主动断开连接，就会进入一个time_wait状态，但是进入这个状态之后，
        //套接字并不会立刻释放，即ip和端口还在被占用，就不能继续使用它们了，这本来是保护客户端的，
        //但是在服务器使用中，如果此时服务器出了问题(比如崩溃，退出了)，就无法立即启动了，所以就要把地址端口重用.
        void ReuseAddress(){
            //setsockopt(int fd, int leve, int optname, void *val, int vallen)
            int val = 1;
            setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR, (void*)&val, sizeof(int));
            val = 1;
            setsockopt(_sockfd, SOL_SOCKET, SO_REUSEPORT, (void*)&val, sizeof(int));
        }
        //设置套接字阻塞属性--设置为非阻塞
        //套接字默认是阻塞操作，当读取完缓冲区里的数据时，程序就被阻塞住了，
        //所以要设置成非阻塞的，才能循环不断的读取数据.
        void NonBlock(){
            //int fcntl(int fd, int cmd, ... /* arg */ );
            int flag = fcntl(_sockfd, F_GETFL, 0); //先获取相关属性
            fcntl(_sockfd, F_SETFL, flag | O_NONBLOCK);
        }
};


class Poller;
class EventLoop;
class Channel
{
    private:
        int _fd;
        EventLoop* _loop;
        uint32_t _events;  //当前正在监控的事件
        uint32_t _revents;  //当前连接触发的事件
        using EventCallback = std::function<void()>;
        EventCallback _read_cb;  //可读事件被触发的回调函数
        EventCallback _write_cb; //可写事件被触发的回调函数
        EventCallback _error_cb; //错误事件被触发的回调函数
        EventCallback _close_cb; //连接断开事件被触发的回调函数
        EventCallback _event_cb; //任意事件被触发的回调函数
    public:
        Channel(EventLoop* loop, int fd):_fd(fd), _events(0), _revents(0), _loop(loop) {}
        int Fd(){ return _fd; }
        uint32_t Events() { return _events; }   //获取想要监控的事件
        void SetREvents(uint32_t events) { _revents = events; } //设置实际就绪的事件
        void SetReadCallback(const EventCallback& cb) { _read_cb = cb; }
        void SetWriteCallback(const EventCallback& cb){ _write_cb = cb; }
        void SetErrorCallback(const EventCallback& cb){ _error_cb = cb; }
        void SetCloseCallback(const EventCallback& cb){ _close_cb = cb; }
        void SetEventCallback(const EventCallback& cb){ _event_cb = cb; }
        //当前是否监控了可读
        bool ReadAble() { return _events & EPOLLIN; }   
        //当前是否监控了可写  
        bool WriteAble(){ return _events & EPOLLOUT; }   
        //启动读事件监控 
        void EnableRead() { _events |= EPOLLIN; Update(); }
        //启动写事件监控   
        void EnableWrite() { _events |= EPOLLOUT; Update(); }  
        //解除读事件监控 
        void DisableRead(){ _events &= ~EPOLLIN; Update(); }     
        //解除写事件监控
        void DisableWrite(){ _events &= ~EPOLLOUT; Update(); }    
        //解除所有事件监控
        void DisableAll() { _events = 0; Update(); }  
        //移除监控
        void Remove() ;     
        void Update() ;
        //事件处理，一旦连接触发了事件，就调用这个函数
        void HandleEvent() {
            if ((_revents & EPOLLIN) || (_revents & EPOLLRDHUP) || (_revents & EPOLLPRI)) {
                /*不管任何事件，都调用的回调函数*/
                if (_read_cb) _read_cb();
            }
            /*有可能会释放连接的操作事件，一次只处理一个，所以使用else if */
            if (_revents & EPOLLOUT) {
                if (_write_cb) _write_cb();
            }else if (_revents & EPOLLERR) {
                //一旦出错，就会释放连接，没必要调用任意回调了，但是可以放在前面调用任意回调
                if (_error_cb) _error_cb();
            }else if (_revents & EPOLLHUP) {
                if (_close_cb) _close_cb();
            }
            if (_event_cb) _event_cb();
        }
};

class Acceptor
{
    private:
        Socket _socket; //用于创建监听套接字
        EventLoop* _loop; //用于对监听套接字进行事件监控
        Channel _channel; //用于对监听套接字进行事件管理
    
        using AcceptCallback = std::function<void(int)>;
        AcceptCallback _accept_cb;
    private:
        /*监听套接字的读事件回调处理函数--获取新连接，调用_accept_cb函数进行新连接处理*/
        void HandleRead(){
            int newfd = _socket.Accept();
            if(newfd < 0) { return; }
            if(_accept_cb) { _accept_cb(newfd); }
        }
        int CreateServer(int port){
            bool ret = _socket.CreateServer(port);
            assert(ret == true);
            return _socket.Fd();
        }
    public:
        /*不能将启动读事件监控，放到构造函数中，必须在设置回调函数后，再去启动*/
        /*否则有可能造成启动监控后，立即有事件，处理的时候，回调函数还没设置：新连接得不到处理，且资源泄漏*/
        Acceptor(EventLoop* loop, int port):_socket(CreateServer(port)), _loop(loop),
            _channel(loop, _socket.Fd()){
                _channel.SetReadCallback(std::bind(&Acceptor::HandleRead, this));
            }
        void SetAcceptCallback(const AcceptCallback &cb) { _accept_cb = cb; }
        void Listen() { _channel.EnableRead(); }
};

#define MAX_EPOLLEVENTS 1024
class Poller
{
    private:
        int _epfd;                                  //epoll的操作句柄
        struct epoll_event _evs[MAX_EPOLLEVENTS];   //epoll_event结构数组，监控是保存所有活跃事件
        //管理描述符与描述符对应的事件管理Channel对象.
        //1.对描述符进行监控，通过Chaannel才能知道描述符需要监控什么事件
        //2.当描述符就绪了，通过描述符在hash表中找到对应的Channel
        std::unordered_map<int, Channel*> _channels; 
    private:
        //对epoll的直接操作
        void Update(Channel *channel, int op){
            //int epoll_ctl(int epfd, int op, int fd, struct epoll_event* ev)
            int fd = channel->Fd();
            struct epoll_event ev;
            ev.data.fd = fd;
            ev.events = channel->Events();
            int ret = epoll_ctl(_epfd, op, fd, &ev);
            if(ret < 0){
                ERR_LOG("epoll_ctl fail!");
            }
            return;
        }
        //判断一个Channel是否已经添加了事件监控
        bool HasChannel(Channel *channel){
            auto it = _channels.find(channel->Fd());
            if(it == _channels.end()){
                return false;
            }
            return true;
        }
    public:
        Poller(){
            _epfd = epoll_create(MAX_EPOLLEVENTS);
            if(_epfd < 0){
                ERR_LOG("epoll_create fail!!");
                abort(); //退出程序
            }
        }
        //添加或修改监控事件
        void UpdateEvent(Channel *channel){
            bool ret = HasChannel(channel);
            if(ret == false){
                //不存在则添加
                _channels.insert({channel->Fd(), channel});
                return Update(channel, EPOLL_CTL_ADD);
            }
            //存在就更新
            return Update(channel, EPOLL_CTL_MOD);
        }
        //移除监控
        void RemoveEvent(Channel *channel){
            auto it = _channels.find(channel->Fd());
            if(it != _channels.end()){
                _channels.erase(it);
            }
            Update(channel, EPOLL_CTL_DEL);
        }
        //开始监控，返回活跃连接
        void Poll(std::vector<Channel*> *active){
            //int epoll_wait(int epfd, struct epoll_event *evs, int maxevents, int timeout)
            int nfds = epoll_wait(_epfd, _evs, MAX_EPOLLEVENTS, -1); //nfds 表示就绪的事件个数，-1 表示阻塞监控
            if(nfds < 0){
                if(errno == EINTR){
                    return;
                }
                ERR_LOG("epoll_wait error: %s\n", strerror(errno));
                abort(); //退出程序
            }
            for(int i = 0; i < nfds; i++){
                auto it = _channels.find(_evs[i].data.fd);
                assert(it != _channels.end());
                it->second->SetREvents(_evs[i].events); //设置实际就绪事件
                active->push_back(it->second);
            }
            return;
        }
};

/*
定时器模块的整合：
timerfd：实现内核每隔一段时间，给进程一次超时事件(timerfd可读)
timerwheel：实现每次执行RuntimeTask, 都可以执行一波到期的定时任务
要实现一个完整的秒级定时器，就要把这两个功能整合到一起：
timerfd设置为每秒钟触发一次定时事件，当事件被触发，则运行一次timerwheel的RuntimeTask，执行一下所有的过期定时任务
*/
using TaskFunc = std::function<void()>;
using ReleaseFunc = std::function<void()>;
//定时器任务类
class TimerTask
{
    private:
        uint64_t _id;         //定时器任务对象id
        uint32_t _timeout;    //定时任务的超时时间
        bool _canceled;       //false-表示定时任务没有被取消，true-表示定时任务被取消了
        TaskFunc _task_cb;    //定时器对象要执行的定时任务
        ReleaseFunc _release; //用于删除TimerWheel中保存的定时器对象信息
    public:
        TimerTask(uint64_t id, uint32_t delay, const TaskFunc& task_cb)
        :_id(id), _timeout(delay), _task_cb(task_cb), _canceled(false)
        {}
        ~TimerTask()
        {
            if(_canceled == false) _task_cb();
            _release();
        }
        void Cancel()
        {
            _canceled = true;
        }
        void SetRelease(const ReleaseFunc& cb)
        {
            _release = cb;
        }
        uint32_t DelayTime()
        {
            return _timeout;
        }
};

//时间轮类的设计
class EventLoop;
class TimerWheel
{
    private:
        using WeakTask = std::weak_ptr<TimerTask>;
        using PtrTask = std::shared_ptr<TimerTask>;
        
        int _tick;         //当前秒针，走到哪了释放到哪里，就相当于执行哪里的任务
        int _capacity;     //表盘最大数量 --- 其实就是最大延迟时间
        std::vector<std::vector<PtrTask>> _wheel;
        std::unordered_map<uint64_t, WeakTask> _timers;

        EventLoop *_loop;
        int _timerfd; //定时器描述符--可读事件回调就是读取计数器，执行定时任务
        std::unique_ptr<Channel> _timer_channel; 
    private:
        void RemoveTimer(uint64_t id)
        {
            auto it = _timers.find(id);
            if(it != _timers.end())
            {
                _timers.erase(id);
            }
        }
        static int CreateTimerfd(){
            //定时器定时的原理：每隔一段时间(定时器的超时时间)，系统就会给这个描述符对应的定时器写入一个8字节数据

            //int timerfd_create(int clockid, int flags);
            //创建定时器
            int timerfd = timerfd_create(CLOCK_MONOTONIC, 0);
            if(timerfd < 0)
            {
                ERR_LOG("timerfd_create fail!");
                abort();
            }
            //启动定时器
            //int timerfd_settime(int fd, int flags, const struct itimerspec *new_value, struct itimerspec *old_value);
            struct itimerspec itime;
            itime.it_value.tv_sec = 1; //第一次超时时间为1s
            itime.it_value.tv_nsec = 0;
            itime.it_interval.tv_sec = 5; //第一次超时后，每次超时的间隔时间
            itime.it_interval.tv_nsec = 0;
            timerfd_settime(timerfd, 0, &itime, nullptr);

            return timerfd;
        }
        int ReadTimerfd(){
            //有可能因为其他描述符的事件处理花费事件比较长，然后在处理定时器描述符事件的时候，有可能就已经超时了很多次
            //read读取到的数据times就是从上一次read之后超时的次数
            uint64_t times; //64个比特位，8字节
            int ret = read(_timerfd, &times, 8);
            if(ret < 0)
            {
                ERR_LOG("ReadTimerfd fail!");
                abort();
            }
            return times;
        }
        //运行定时任务
        //这个函数每秒执行一次，相当于秒针向后走一步
        void RunTimerTask()
        {
            _tick = (_tick + 1) % _capacity;
            _wheel[_tick].clear(); //清空指定位置的数组，就会把数组中保存的所有管理定时对象的shared_ptr释放掉
        }
        //时间到了
        void OnTime(){
            //根据实际超时的次数，执行对应的超时任务
            int times = ReadTimerfd();
            for(int i = 0; i < times; i++){
                RunTimerTask();
            }
        }
        //添加定时任务
         void TimerAddInLoop(uint64_t id, uint32_t delay, const TaskFunc& cb)
        {
            PtrTask pt(new TimerTask(id, delay, cb)); //实例化任务对象
            pt->SetRelease(std::bind(&TimerWheel::RemoveTimer, this, id)); //当这个定时任务被销毁的时候，需要从unordered_map中把信息移除掉
            int pos = (_tick + delay) % _capacity; //计算出添加的位置.表盘是一个环，要回绕
            _wheel[pos].push_back(pt);             //把任务添加进表盘位置
            _timers[id] = WeakTask(pt);

        }
        //刷新/延迟定时任务
        void TimerRefreshInLoop(uint64_t id)
        {
            //通过保存的定时器对象的weak_ptr,构造出一个sheard_ptr出来， 添加进轮子中
            auto it = _timers.find(id);
            if(it == _timers.end()){
                return; //没找到定时任务，没法刷新
            }
            //lock:获取weak_ptr管理的对象对应的shared_ptr
            //若任务对象被销毁，则返回空的 shared_ptr，
            //如果存在，则返回一个 shared_ptr，增加引用计数，确保对象在使用期间不会被销毁.
            PtrTask pt = it->second.lock();
            int delay = pt->DelayTime();
            int pos = (_tick + delay) % _capacity; //计算出添加的位置.表盘是一个环，要回绕
            _wheel[pos].push_back(pt);             //把任务添加进表盘位置
        }
        //取消定时任务
        void TimerCancelInLoop(uint64_t id)
        {
            auto it = _timers.find(id);
            if(it == _timers.end())
            {
                return; //没找到定时任务，没法刷新
            }
            PtrTask pt = it->second.lock();
            if(pt) pt->Cancel();
        }
    public:
        //初始化不取决于初始化列表的顺序，取决于定义的顺序
        TimerWheel(EventLoop* loop)
        :_capacity(60), 
        _tick(0), 
        _wheel(_capacity),
        _loop(loop),
        _timerfd(CreateTimerfd()),
        _timer_channel(new Channel(_loop, _timerfd))
        {
            _timer_channel->SetReadCallback(std::bind(&TimerWheel::OnTime, this));
            _timer_channel->EnableRead(); //启动读事件监控
        }
        //添加定时任务
        //定时器中有个_timers成员，定时器信息的操作有可能在多线程中进行，因此需要考虑线程安全问题
        //如果不想加锁，那就把定时器的所有操作，都放到同一个线程中执行
        void TimerAdd(uint64_t id, uint32_t delay, const TaskFunc& cb);
        //刷新/延迟定时任务
        void TimerRefresh(uint64_t id);
        //取消定时任务
        void TimerCancel(uint64_t id) ;
        
        /*这个接口存在线程安全问题*/
        bool HasTimer(uint64_t id){
            auto it = _timers.find(id);
            if(it == _timers.end()){
                return false; 
            }
            return true;
        }
};


/*
完成EventLoop模块前置知识：
eventfd:一种事件通知机制。创建一个描述符用于事件通知
eventfd的本质是在内核中管理的一个计数器
创建eventfd就会在内核中创建计数器(结构)
每当向eventfd中写入一个值 -- 用于表示事件通知次数
可以使用read进行数据的读取，读取到的数据就是通知的次数

用处：在EventLoop模块中实现线程间的事件通知功能。

eventfd也是通过read/write/close进行操作的
注意点：使用 read/write 进行IO的时候，数据只能是一个8字节数据
*/

/*
EventLoop模块：进行事件监控，以及事件处理的模块
关键点：这个模块与线程是一一对应关联的。

监控了一个连接，而这个连接一旦就绪，就要进行事件处理。
但是如果这个描述符在多个线程中都触发的事件，进行处理，就会存在线程安全问题，
因此我们需要将一个连接的事件监控，以及连接事件处理，以及其他操作都放在同一个线程中进行。

如何保证一个连接的所有操作都在EventLoop对应的线程中？
解决方案：给EventLoop模块中，添加一个任务队列，对连接的所有操作，都进行一次封装，
        将对连接的操作并不直接执行，而是当作任务添加到任务队列中。

EventLoop处理流程：
1.在线程中对描述符进行事件监控
2.有描述符就绪就对描述符进行事件处理(如何保证处理回调函数中的操作都在线程中?)
3.所有的就绪事件处理完了，这时候在去将任务队列中的所有任务一一执行

再次进行理解：
EventLoop中有两个关键的东西：epoll，用于对描述符的监控; task，里面有对连接的所有操作的封装
当epoll中有描述符就绪了，就要进行事件的处理--调用回调函数，
假设在处理过程中调用了send，而这个send是封装后的send，实际上内部是将这个数据发送的操作，压入任务队列task,
等到所有的就绪事件都处理完了，然后从task任务队列中一一取出实际要进行操作的执行。
这样我们只需要把任务队列加一把锁，就不涉及线程安全问题了。

当事件就绪，需要处理的时候，处理过程中，如果对连接要进行某些操作，
这些操作必须在EventLoop对应的线程中执行，保证对连接的各项操作都是线程安全的。
1.如果执行的操作本就在线程中，不需要将操作压入队列了，可以直接执行
2.如果执行的操作不在线程中，才需要加入任务队列，等到事件处理完了任何才执行任务
所以需要有 std::thread::id _id 来判断当前要进行的操作是不是在它对应保存的那个线程中。
*/
class EventLoop
{
    private:
        std::thread::id _thread_id; //线程id
        int _event_fd; //eventfd唤醒IO事件有可能导致的阻塞
        std::unique_ptr<Channel> _event_channel;
        Poller _poller; //进行所有描述符的事件监控

        using Functor = std::function<void()>;
        std::vector<Functor> _task; //任务池(任务队列)
        std::mutex _mutex; //互斥锁，保证任务池操作的线程安全

        TimerWheel _timer_wheel; //定时器模块
    public:
        //运行任务池中的所有任务
        void RunAllTask(){
            std::vector<Functor> functor;
            {
                std::unique_lock<std::mutex> _lock(_mutex); //出了这个{}锁就会释放
                _task.swap(functor); //把任务队列中的所有任务都交换出了
            }
            //一一取出来执行
            for(auto& f : functor){
                f();
            }
            return;
        }
        static int CreateEventFd(){
            int efd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
            if(efd < 0){
                ERR_LOG("create eventfd fail!");
                abort();
            }
            return efd;
        }
        void ReadEventfd(){
            uint64_t res = 0;
            int ret = read(_event_fd, &res, sizeof(res));
            if(ret < 0){
                if(errno == EINTR || errno == EAGAIN){
                    return;
                }
                ERR_LOG("read _event_fd fail!");
                abort();
            }
            return;
        }
        void WeakUpEventfd(){
            uint64_t val = 1;
            int ret = write(_event_fd, &val, sizeof(val));
            if(ret < 0){
                if(errno == EINTR){
                    return;
                }
                ERR_LOG("write _event_fd fail!");
                abort();
            }
            return;
        }
    public:
        EventLoop()
        :_thread_id(std::this_thread::get_id()),
         _event_fd(CreateEventFd()),
         _event_channel(new Channel(this, _event_fd)),
         _timer_wheel(this)
         {
            //给eventfd添加可读事件回调函数，读取eventfd事件通知次数
            _event_channel->SetReadCallback(std::bind(&EventLoop::ReadEventfd, this));
            //启动eventfd的读事件监控
            _event_channel->EnableRead();
         }
         //启动EventLoop模块
        //事件监控 --> 就绪事件处理 --> 执行任务
        void Start(){
            while(1){
                //1.事件监控
                std::vector<Channel*> actives;
                _poller.Poll(&actives);
                //2.事件处理
                for(auto& channel : actives){
                    channel->HandleEvent();
                }
                //3.执行任务
                RunAllTask();
            }
        }
        //用于判断当前线程是否是EventLoop对应的线程,即让_id与对应线程id进行比较
        bool IsInLoop(){
            //std::this_thread::get_id() 当前运行线程的id
            //_thread_id EventLoop实例化时所关联线程的id
            return (_thread_id == std::this_thread::get_id());
        }
        void AssertInLoop(){
            assert(_thread_id == std::this_thread::get_id());
        }
        //判断将要执行的任务是否处于当前线程中，如果是则执行，不是则压入队列
        void RunInLoop(const Functor& cb){
            if(IsInLoop()){
                return cb();
            }
            return QueueInLoop(cb);
        }
        //将操作压入任务池
        void QueueInLoop(const Functor& cb){
            {
                std::unique_lock<std::mutex> _lock(_mutex);
                _task.push_back(cb);
            }
            //唤醒有可能没有事件就绪，而导致的epoll阻塞
            //其实就是给eventfd写入一个数据，eventfd就会触发就绪事件
            WeakUpEventfd();
        }
       
        //添加/修改描述符的监控事件
        void UpdateEvent(Channel *channel){
            return _poller.UpdateEvent(channel);
        }
        //移除描述符的监控事件
        void RemoveEvent(Channel *channel){
            return _poller.RemoveEvent(channel);
        }
        //添加定时器任务
        void TimerAdd(uint64_t id, uint32_t delay, const TaskFunc& cb){
            return _timer_wheel.TimerAdd(id, delay, cb);
        }
        //刷新定时任务
        void TimerRefresh(uint64_t id){
            return _timer_wheel.TimerRefresh(id);
        }
        //取消定时任务
        void TimerCancel(uint64_t id){
            _timer_wheel.TimerCancel(id);
        }
        //是否有某个定时器任务存在
        bool HasTimer(uint64_t id){
            return _timer_wheel.HasTimer(id);
        }
        
};

/*LoopThread模块：
功能：将EventLoop与thread整合到一起，可以向外返回所实例化的EventLoop对象。
含义：EventLoop模块在实例化对象的时候，必须在线程内部。
    EventLoop在实例化对象时会设置自己的thread_id,如果我们先创建了多个EventLoop对象，然后创建了多个线程，再将各个线程的id，
    重新给EventLoop进行设置，会存在问题：在构造EventLoop对象，到设置新的thread_id期间将是不可控的。
    因此我们必须先创建线程，然后在线程的入口函数中，去实例化EventLoop对象。
*/
class LoopThread
{
    private:
        /*_mutex和_cond用于实现_loop获取的同步关系，避免线程创建了，但是_loop还没有实例化之前去获取_loop*/
        std::mutex _mutex;          //互斥锁
        std::condition_variable _cond;   //条件变量
        EventLoop* _loop;       //EventLoop的指针变量，这个对象需要在线程内实例化
        std::thread _thread;    //EventLoop对应的线程
    private:
        /*实例化 EventLoop 对象，唤醒_cond上有可能阻塞的线程，并且开始运行EventLoop模块的功能*/
        void ThreadEntry(){
            EventLoop loop;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _loop = &loop;
                _cond.notify_all();
            }
            loop.Start();
        }
    public:
        /*创建线程，设定入口函数*/
        LoopThread():_loop(NULL), _thread(std::thread(&LoopThread::ThreadEntry, this)){}
        /*返回当前线程关联的EventLoop对象指针*/
        EventLoop *GetLoop(){
            EventLoop *loop = NULL;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                //满足_loop != NULL，则不等待了
                _cond.wait(lock, [&](){ return _loop != NULL; });
                loop = _loop;
            }
            return loop;
        }
};

class LoopThreadPool 
{
    private:
        int _thread_count;    //从属线程的数量
        int _next_idx;        //_loop是下标id
        EventLoop *_baseloop; //主EventLoop，运行在主线程，当从属线程数量为0时，所有操作都在_baseloop中进行
        std::vector<LoopThread*> _threads; //保存所有LoopThread对象
        std::vector<EventLoop *> _loops;   //当从属线程数量大于0，则从_loops中进行线程EventLoop的分配
    public:
        LoopThreadPool(EventLoop *baseloop):_thread_count(0), _next_idx(0), _baseloop(baseloop) {}
        //设置线程数量
        void SetThreadCount(int count) { _thread_count = count; }
        //创建所有的从属线程
        void Create() {
            if (_thread_count > 0) {
                _threads.resize(_thread_count);
                _loops.resize(_thread_count);
                for (int i = 0; i < _thread_count; i++) {
                    _threads[i] = new LoopThread();
                    _loops[i] = _threads[i]->GetLoop();
                }
            }
            return ;
        }
        EventLoop *NextLoop() {
            //当从属线程数量为0时，所有操作都在_baseloop中进行
            if (_thread_count == 0) {
                return _baseloop;
            }
            _next_idx = (_next_idx + 1) % _thread_count;
            return _loops[_next_idx];
        }
};

class Any
{
    private:
        class holder //父类
        {
            public:
                virtual ~holder() {}
                virtual const std::type_info& type() = 0; 
                virtual holder *clone() = 0; 
        };
        template<class T>
        class placeholder:public holder //子类
        {
            public:
                placeholder(const T& val) :_val(val) {}
                //获取子类当前的数据类型T
                virtual const std::type_info& type() { return typeid(T); } 
                //针对当前对象自身克隆出新的子类对象出来
                virtual holder *clone() { return new placeholder(_val); }
            public:
                T _val; //保存任意类型的数据
        };
        holder *_content;
    public:
        Any(): _content(nullptr) {}; //构造空容器

        template<class T>
        //构造任意类型的通用容器.我们并不知道类型是什么，设置成模板函数
        Any(const T &val): _content(new placeholder<T>(val)) {}; 
        //根据其他的容器构造新容器
        //不是保存对方容器数据的指针，而是容器中的数据
        Any(const Any &other): _content(other._content ? other._content->clone() : nullptr){}; 

        //销毁_content
        ~Any() {delete _content; } 

        template<class T>
        //获取子类对象当中保存数据的指针 &val
        T *get() 
        {
            //想要获取的数据类型，必须和保存的数据类型一致
            assert(typeid(T) == _content->type());
            //注意优先级：先强转，再取出_val，最后取_val的地址
            return &((placeholder<T>*)_content)->_val;
        }

        Any& swap(Any& other)
        {
            std::swap(_content, other._content);
            return *this;
        }

        template<class T>
        //赋值运算符重载 - 赋值各种类型的数据
        Any& operator=(const T& val) 
        {
            //为val构造一个临时通用容器，然后与当前容器自身进行指针交换，临时对象释放3的时候，
            //原先保存的数据也被释放了
            Any(val).swap(*this);
            return *this;
        } 
        //赋值运算符重载 - 赋值各种类型的通用容器
        Any& operator=(const Any& other) 
        {
            Any(other).swap(*this);
            return *this;
        } 
};


class Connection;
//DISCONNECTED -- 连接关闭状态        CONNECTING -- 连接建立成功，待处理状态
//CONNECTED -- 连接建立完成，各种设置已完成，可以通信状态     DISCONNECTING -- 待关闭状态
typedef enum { DISCONNECTED, CONNECTING, CONNECTED, DISCONNECTING} ConnStatu;
using PtrConnection = std::shared_ptr<Connection>;
class Connection : public std::enable_shared_from_this<Connection>
{
    private:
        uint64_t _conn_id;      //连接的唯一id,便于连接的管理和查找
        //uint64_t _timer_id;   定时器id,必须是唯一的，这里为了简化操作使用_conn_id作为定时器id
        int _sockfd;            //连接关联的文件描述符
        bool _enable_inactive_release;  //连接是否启动非活跃销毁的标志，默认为false
        EventLoop* _loop;       //连接所关联的EventLoop
        ConnStatu _statu;       //连接的状态
        Socket _socket;         //套接字操作管理
        Channel _channel;       //连接事件管理
        Buffer _in_buffer;      //输入缓冲区--存放从socket中读取到的数据
        Buffer _out_buffer;     //输出缓冲区--存放要发送给对端的数据
        Any _context;            //请求的接收处理上下文

        /*这4个回调函数，是让组件使用者来设置的*/
        using ConnectedCallback = std::function<void(const PtrConnection&)>;
        using MessageCallback = std::function<void(const PtrConnection&, Buffer* )>;
        using ClosedCallback = std::function<void(const PtrConnection&)>;
        using AnyEventCallback = std::function<void(const PtrConnection&)>;
        ConnectedCallback _connected_cb;
        MessageCallback _message_cb;
        ClosedCallback _closed_cb;
        AnyEventCallback _event_cb;
        //组件内的连接关闭回调--组件内设置的，因为服务器组件内会把所有的连接管理起来
        //一旦某个连接要关闭，就应该从管理的地方移除掉自己的信息
        ClosedCallback _server_closed_cb;
    private:
        /*5个channel的事件回调函数*/
        //描述符触发可读事件后调用的函数，接收socket数据放到接收缓冲区中，然后调用_message_cb
        void HandleRead(){
            //1.接收socket的数据
            char buf[65535];
            ssize_t ret = _socket.NonBlockRecv(buf, 65535);
            if(ret < 0){
                //出错了，不能直接关闭连接
                return ShutdownInLoop();
            }
            //这里如果ret等于0表示没有读取到数据，而并不是断开连接，断开连接返回的是-1
            //将数据放入输入缓冲区,写入之后顺便将写偏移向后移动
            _in_buffer.WriteAndPush(buf, ret);
            //2.调用_message_cb进行业务处理
            if(_in_buffer.ReadAbleSize() > 0){
                //sharee_from_this 从当前对象自身获取自身的shared_ptr管理对象
                return _message_cb(shared_from_this(), &_in_buffer);
            }
        }
        //描述符触发可写事件后调用的函数，将发送缓冲区的数据进行发送
        void  HandleWrite(){
            //_out_buffer中保存的数据就是要发送的数据
            ssize_t ret = _socket.NonBlockSend(_out_buffer.ReadPos(), _out_buffer.ReadAbleSize());
            if(ret < 0){
                if(_in_buffer.ReadAbleSize() > 0){
                    //sharee_from_this 从当前对象自身获取自身的shared_ptr管理对象
                    _message_cb(shared_from_this(), &_in_buffer);
                }
                return ReleaseInLoop(); //这时候就是实际的关闭释放操作了
            }
            _out_buffer.MoveReadOffset(ret); //千万不用忘了，将读偏移向右移动
            //发送缓冲区数据为0，就直接关闭写事件监控了，避免不断的触发写事件但是又没有数据发送
            if(_out_buffer.ReadAbleSize() == 0){
                _channel.DisableWrite();
                //如果当前是连接待关闭状态，若有数据，发送完数据释放连接，没有数据则直接释放
                if(_statu == DISCONNECTING && _out_buffer.ReadAbleSize() == 0){
                    return Release();
                }
            }
            
            return;
        }
        //描述符触发挂断事件后调用的函数
        void  HandleClose(){
            /*一旦连接挂断了，套接字就什么都不干了，因此有数据待处理就处理一下，完毕关闭连接*/
            if(_in_buffer.ReadAbleSize() > 0){
                _message_cb(shared_from_this(), &_in_buffer);
            }
            return Release(); //这时候就是实际的关闭释放操作了
        }
        //描述符触发错误事件后调用的函数
        void  HandleError(){
            return HandleClose();
        }
        //描述符触发任意事件后调用的函数:
        //1.刷新连接的活跃度--延迟定时销毁任务
        //2.调用组件使用者的任意事件回调
        void  HandleEvent(){
            if(_enable_inactive_release == true){_loop->TimerRefresh(_conn_id); }
            if(_event_cb){ _event_cb(shared_from_this()); }
        }

        //下面这些操作都要放在EventLoop中的任务队列里执行的：
        //连接完成后，所处的状态下要进行各种设置(启动读监控,调用回调函数)
        void EstablishedInLoop(){
            //1.修改连接状态  2.启动读事件监控  3.调用回调函数
            assert(_statu == CONNECTING); //当前状态必须一定是上层的半连接状态
            _statu == CONNECTED; //当前函数执行完毕，则连接进入已完成连接状态
            //一旦启动读事件监控就有可能会立即出发读事件，如果这时候启动了非活跃连接销毁
            _channel.EnableRead();
            if(_connected_cb) _connected_cb(shared_from_this());
        }
        //实际的释放接口
        void ReleaseInLoop(){
            //1.修改连接状态
            _statu == DISCONNECTED;
            //2.移除连接的事件监控
            _channel.Remove();
            //3.关闭描述符
            _socket.Close();
            //4.如果当前定时器队列中还有定时销毁任务，则取消任务
            if(_loop->HasTimer(_conn_id)) CancelInactiveReleaseInLoop();
            //5.调用关闭回调函数，避免先移除服务器管理的连接信息导致Connection被释放，再去处理会出错，因此先调用用户的
            if(_closed_cb) _closed_cb(shared_from_this());
            //移除服务器内部管理的连接信息
            if(_server_closed_cb) _server_closed_cb(shared_from_this());
        }
        //这个接口并不是实际的发送接口，而只是把数据放到了发送缓冲区，启动了可写事件监控
        void SendInLoop(Buffer &buf) {
            if (_statu == DISCONNECTED) return ;
            _out_buffer.WriteBufferAndPush(buf);
            if (_channel.WriteAble() == false) {
                _channel.EnableWrite();
            }
        }
        //这个关闭操作并非实际的连接释放操作，需要判断还有没有数据待处理，待发送
        void ShutdownInLoop(){
            _statu = DISCONNECTING; //设置为半关闭状态
            if(_in_buffer.ReadAbleSize() > 0){
                if(_message_cb) _message_cb(shared_from_this(), &_in_buffer);
            }
            //要么就是写入数据的时候出错关闭，要么就是没有待发送数据，直接关闭
            if(_out_buffer.ReadAbleSize() > 0){
                if(_channel.WriteAble() == false){
                    _channel.EnableWrite();
                }
            }
            if(_out_buffer.ReadAbleSize() == 0){
                Release();
            }
        }
        //启动非活跃连接超时释放规则
        void EnableInactiveReleaseInLoop(int sec){
            //1.将判断标志_enable_inactive_release置为true
            _enable_inactive_release = true;
            //2.如果当前定时销毁任务已经存在，那就刷新延迟一下
            if(_loop->HasTimer(_conn_id)){
                return _loop->TimerRefresh(_conn_id);
            }
            //3.如果不存在定时销毁任务，则新增
            _loop->TimerAdd(_conn_id, sec, std::bind(&Connection::Release, this));
        }
        //取消非活跃连接超时释放规则
        void CancelInactiveReleaseInLoop(){
            _enable_inactive_release = false;
            if(_loop->HasTimer(_conn_id)){
                _loop->TimerCancel(_conn_id);
            }
        }
        //升级/切换协议
        void UpgradeInLoop(const Any &context, 
                    const ConnectedCallback& conn, 
                    const MessageCallback& msg, 
                    const ClosedCallback& closed, 
                    const AnyEventCallback& event){
                        _context = context;
                        _connected_cb = conn;
                        _message_cb = msg;
                        _closed_cb = closed;
                        _event_cb = event;
                    }    
    public:
       Connection(EventLoop *loop, uint64_t conn_id, int sockfd):_conn_id(conn_id), _sockfd(sockfd),
            _enable_inactive_release(false), _loop(loop), _statu(CONNECTING), _socket(_sockfd),
            _channel(loop, _sockfd) {
            _channel.SetCloseCallback(std::bind(&Connection::HandleClose, this));
            _channel.SetEventCallback(std::bind(&Connection::HandleEvent, this));
            _channel.SetReadCallback(std::bind(&Connection::HandleRead, this));
            _channel.SetWriteCallback(std::bind(&Connection::HandleWrite, this));
            _channel.SetErrorCallback(std::bind(&Connection::HandleError, this));
        }
        ~Connection() { DBG_LOG("release connection: %p", this); }
        //获取文件描述符
        int Fd() { return _sockfd; }
        //获取连接Id
        int Id() { return _conn_id; }
        //判断是否处于CONNECTED连接完成状态
        bool Connected() { return (_statu == CONNECTED); }
        //设置上下文--连接建立完成时调用
        void SetContext(const Any& context) { _context = context; }
        //获取上下文--返回的是指针，要保证我们访问的是同一个数据
        Any *GetContext() { return &_context; }
        void SetConnectedCallback(const ConnectedCallback& cb) { _connected_cb = cb; }
        void SetMessageCallback(const MessageCallback& cb) { _message_cb = cb; }
        void SetClosedCallback(const ClosedCallback& cb) { _closed_cb = cb ;}
        void SetAnyEventCallback(const AnyEventCallback& cb) { _event_cb = cb; }
        void SetSrvClosedCallback(const AnyEventCallback& cb) { _server_closed_cb = cb; }
        //发送数据，将数据放到发生缓冲区，启动写事件监控
        void Send(const char* date, size_t len){
            //外界传入的data，可能是个临时的空间，我们现在只是把发送操作压入了任务池，有可能并没有被立即执行
            //因此有可能执行的时候，data指向的空间有可能已经被释放了。
            Buffer buf;
            buf.WriteAndPush((void*)date, len);
            _loop->RunInLoop(std::bind(&Connection::SendInLoop, this, std::move(buf)));
        }
        //连接建立就绪后，给channel设置事件回调，启动读监控，调用_connected_cb
        void Established(){
            _loop->RunInLoop(std::bind(&Connection::EstablishedInLoop, this));
        }
         void Release(){
            _loop->QueueInLoop(std::bind(&Connection::ReleaseInLoop, this));
        }
        //提供给组件使用者的关闭接口--并不实际关闭，需要判断有没有数据待处理
        void Shutdown(){
            _loop->RunInLoop(std::bind(&Connection::ShutdownInLoop, this));
        }
        //启动非活跃连接的销毁,并定义多长时间无通信就是非活跃，添加定时任务
        void EnableInactiveRelease(int sec){
            _loop->RunInLoop(std::bind(&Connection::EnableInactiveReleaseInLoop, this, sec));
        }
        //取消非活跃连接的销毁
        void CancelInactiveRelease(){
            _loop->RunInLoop(std::bind(&Connection::CancelInactiveReleaseInLoop, this));
        }
        //协议切换，重置上下文以及阶段性处理函数
        //协议切换其实就是切换的数据的处理方式，所以要把各种函数和上下文切换掉.
        //这个接口必须在EventLoop线程中立即执行，防备新的事件触发后，处理的时候，切换任务没有被执行，
        //那就会导致数据使用原协议处理了
        void Upgrade(const Any &context, const ConnectedCallback& conn, const MessageCallback& msg, 
                    const ClosedCallback& closed, const AnyEventCallback& event){
                        _loop->AssertInLoop();
                        _loop->RunInLoop(std::bind(&Connection::UpgradeInLoop, this, context, conn, msg, closed, event));
                    }


};

class TcpServer
{
    private:
        uint64_t _next_id;     //这是一个自动增长的连接id
        int _port;
        int _timeout;          //非活跃连接的统计时间--多长时间无通信就是非活跃连接
        bool _enable_inactive_release; //是否启动了非活跃连接超时销毁的判断标志
        EventLoop _baseloop;  //这是主线程的EventLoop对象，负责监听事件的处理
        Acceptor _acceptor;    //这是监听套接字的管理对象
        LoopThreadPool _pool;  //这是从属EventLoop线程池
        std::unordered_map<uint64_t, PtrConnection> _conns; //保存管理所有连接对应的shared_ptr对象

        using Functor = std::function<void()>;

        using ConnectedCallback = std::function<void(const PtrConnection&)>;
        using MessageCallback = std::function<void(const PtrConnection&, Buffer* )>;
        using ClosedCallback = std::function<void(const PtrConnection&)>;
        using AnyEventCallback = std::function<void(const PtrConnection&)>;
        ConnectedCallback _connected_cb;
        MessageCallback _message_cb;
        ClosedCallback _closed_cb;
        AnyEventCallback _event_cb;
    private:
        void RunAfterInLoop(const Functor& task, int delay){
            _next_id++;
            _baseloop.TimerAdd(_next_id, delay, task);
        }
        //为新连接构造一个connection进行管理
        void NewConnection(int fd){
            //DBG_LOG("NewConnection...");
            _next_id++;
            PtrConnection conn(new Connection(_pool.NextLoop(), _next_id, fd));
            conn->SetMessageCallback(_message_cb); 
            conn->SetClosedCallback(_closed_cb); 
            conn->SetConnectedCallback(_connected_cb); 
            conn->SetAnyEventCallback(_event_cb);
            conn->SetSrvClosedCallback(std::bind(&TcpServer::RemoveConnection, this, std::placeholders::_1));
            if(_enable_inactive_release) conn->EnableInactiveRelease(_timeout); //启动非活跃超时销毁功能
            conn->Established(); //就绪初始化
            _conns.insert({_next_id, conn}); 
        }
        //从管理connection中移除连接信息
        void RemoveConnection(const PtrConnection &conn){
            int id = conn->Id();
            auto it = _conns.find(id);
            if (it != _conns.end()) {
                _conns.erase(it);
            }
        }
    public:
        TcpServer(int port):
        _port(port),
        _next_id(0),
        _enable_inactive_release(false),
        _acceptor(&_baseloop, port),
        _pool(&_baseloop){
            _acceptor.SetAcceptCallback(std::bind(&TcpServer::NewConnection, this, std::placeholders::_1));
            _acceptor.Listen(); //将监听套接字挂到baseloop上开始监控事件
        } 
        void SetThreadCount(int count){
            return _pool.SetThreadCount(count);
        }
        void SetConnectedCallback(const ConnectedCallback& cb) { _connected_cb = cb; }
        void SetMessageCallback(const MessageCallback& cb) { _message_cb = cb; }
        void SetClosedCallback(const ClosedCallback& cb) { _closed_cb = cb ;}
        void SetAnyEventCallback(const AnyEventCallback& cb) { _event_cb = cb; }
        //是否启动超时销毁
        void EnableInactiveRelease(int timeout){
            _timeout = timeout;
            _enable_inactive_release = true;
        }
        //用于添加一个定时任务--多少秒之后执行一个定时任务
        void RunAfter(const Functor& task, int delay){
            _baseloop.RunInLoop(std::bind(&TcpServer::RunAfterInLoop,this, task, delay));
        }
        void Start(){
            _pool.Create(); //创建线程池中的从属线程
            _baseloop.Start();
        }
};


void Channel::Remove() {  return _loop->RemoveEvent(this); }     
void Channel::Update() {  return _loop->UpdateEvent(this); }

//添加定时任务
//定时器中有个_timers成员，定时器信息的操作有可能在多线程中进行，因此需要考虑线程安全问题
//如果不想加锁，那就把定时器的所有操作，都放到同一个线程中执行
void  TimerWheel::TimerAdd(uint64_t id, uint32_t delay, const TaskFunc& cb){
     _loop->RunInLoop(std::bind(&TimerWheel::TimerAddInLoop, this, id, delay, cb));
}
//刷新/延迟定时任务
void TimerWheel::TimerRefresh(uint64_t id){
    _loop->RunInLoop(std::bind(&TimerWheel::TimerRefreshInLoop, this, id));
}
//取消定时任务
void TimerWheel::TimerCancel(uint64_t id){
    _loop->RunInLoop(std::bind(&TimerWheel::TimerCancelInLoop, this, id));
}

class NetWork {
    public:
        NetWork() {
            DBG_LOG("SIGPIPE INIT");
            signal(SIGPIPE, SIG_IGN);
        }
};
static NetWork nw;


